package org.niit.handler

import com.google.gson.Gson
import org.apache.kafka.clients.consumer.ConsumerRecord
import org.apache.spark.streaming.dstream.{DStream, InputDStream}
import org.niit.bean.{AdClickData, Answer}
import org.niit.util.{MyKafkaUtil, SparkUtil}

object DataHandler {

  private val ssc = SparkUtil.takeSSC()
  //处理Kafka的数据       将组名和主题传递进来
  def KafkaDataHandler(groupId:String,topic:String): DStream[AdClickData] ={
    val kfDataDS: InputDStream[ConsumerRecord[String, String]] = MyKafkaUtil.getKafkaStream(groupId,topic, ssc)
    val adClickData: DStream[AdClickData] = kfDataDS.map(kafkaData => {
      val data = kafkaData.value()
      val datas = data.split(" ")
      AdClickData(datas(0), datas(1), datas(2), datas(3), datas(4))
    })
    adClickData
  }

  def kafkaAnswerDataHandler(groupId:String,topic:String):DStream[Answer]={

    val kfDataDS: InputDStream[ConsumerRecord[String, String]] = MyKafkaUtil.getKafkaStream(groupId,topic, ssc)
    val answerData: DStream[Answer] = kfDataDS.map(kafkaData => {
      val data:String = kafkaData.value()
      val gson = new Gson();//Gson 是将JSON格式的字符串转换scala样例类的工具，也可以对象转成JSON格式的字符串
      /*
      JS中对象  {
        "student_id":"学生ID_31",
        "textbook_id":"教材ID_1",
        "grade_id":"年级ID_6",
        "subject_id":"科目ID_2_语文",
        "chapter_id":"章节ID_chapter_3",
        "question_id":"题目ID_1003",
        "score":7,
        "answer_time":"2021-01-09 14:53:28",
        "ts":"Jan 9, 2021 2:53:28 PM"
      }
      Java中的对象  ----- 由类实例化
      JavaScript中对象 ----- 由JSON来的
      前端  -->  JSON    ——> 后端
            {                                                     User.class
              "username":"zhangsan",       GSON                         String userName = "zhangsan"
              "password":"123456"                                       String password
            }
            ORM:关系型映射  ----> 应用于数据库和 对象  数据库中的字段名 和 对象名一致
            user.getUsername()         约定 优于 编码
       */

      val answer: Answer = gson.fromJson(data, classOf[Answer])
      answer
    })
    answerData
  }


  //让ssc开启采集和等待采集（阻塞主线程）
  def startAndAwait(): Unit ={
    ssc.start()
    ssc.awaitTermination()
  }

}
